package com.rem.flink.flink5Watermark;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 从输入端获取
 * 无界流处理  针对有序流 forMonotonousTimestamps
 *
 * @author Rem
 * @date 2022-09-26
 */

public class WaterMarkMonotonousTest {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.从socket中读取流
        DataStreamSource<Event> stream = streamEnv.addSource(new ClickSource());

        // 插入水位线的逻辑
        stream.assignTimestampsAndWatermarks(
                        //插入水位线 针对乱序设置延迟时间  顺序不需要
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                //提取时间戳 取事件事件
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()))
                .keyBy(Event::getUser)
                //设置滚动事件窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("count")
                //.reduce((a, b) -> a.setCount(a.getCount() + b.getCount()))
                .print();


        //7.执行流
        streamEnv.execute();

    }
}
